package com.xjaq.dify.service.impl;

import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.xjaq.dify.common.Result;
import com.xjaq.dify.config.DiFyConfig;
import com.xjaq.dify.entity.*;
import com.xjaq.dify.http.AudioApi;
import com.xjaq.dify.http.HttpApi;
import com.xjaq.dify.http.UploadHttpApi;
import com.xjaq.dify.service.IPsyAiChatService;
import lombok.extern.slf4j.Slf4j;
import okhttp3.MultipartBody;
import okhttp3.RequestBody;
import okhttp3.ResponseBody;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.web.multipart.MultipartFile;
import org.springframework.web.servlet.mvc.method.annotation.ResponseBodyEmitter;
import retrofit2.Call;
import retrofit2.Response;

import java.io.*;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@Slf4j
@Service
public class IPsyAiChatServiceImpl implements IPsyAiChatService {

    @Autowired
    private DiFyConfig diFyConfig;
    @Autowired
    private HttpApi httpApi;
    @Autowired
    private UploadHttpApi uploadHttpApi;
    @Autowired
    private AudioApi audioApi;
    private final ObjectMapper  objectMapper = new ObjectMapper();
    private final ExecutorService executorService = Executors.newFixedThreadPool(10);
    @Override
    public ResponseBodyEmitter sseChatPrompt(DifyRequest difyRequest) {
        ResponseBodyEmitter emitter = new ResponseBodyEmitter(diFyConfig.getEmitterTimeout());
        try {
                // 发送请求到 Dify API
//                sendDifyRequest(difyRequest, emitter);
            Call<ResponseBody> responseBodyCall = httpApi.chatMessages(difyRequest);

            this.enqueue(responseBodyCall,emitter);
//                processStreamResponse(, emitter);
//                log.info("成功!");
            } catch (Exception e) {
                log.error("调用 Dify API 失败", e);
                try {
                    emitter.send("Error: " + e.getMessage());
                } catch (IOException ioException) {
                    log.error("发送错误信息失败", ioException);
                } finally {
                    emitter.complete();
                }
            }
        return emitter;
    }

    @Override
    public ResponseBodyEmitter uploadPrompt(DifyRequest difyRequest) {
        ResponseBodyEmitter emitter = new ResponseBodyEmitter(diFyConfig.getEmitterTimeout());
        try {
            // 发送请求到 Dify API
//                sendDifyRequest(difyRequest, emitter);
            Call<ResponseBody> responseBodyCall = uploadHttpApi.uploadFile(difyRequest);

            this.enqueue(responseBodyCall,emitter);
//                processStreamResponse(, emitter);
//                log.info("成功!");
        } catch (Exception e) {
            log.error("调用 Dify API 失败", e);
            try {
                emitter.send("Error: " + e.getMessage());
            } catch (IOException ioException) {
                log.error("发送错误信息失败", ioException);
            } finally {
                emitter.complete();
            }
        }

        return emitter;
    }

    @Override
    public Result stopChatTask(String taskId, String user) {
        try {
            // 构造请求体
            Map<String, String> requestBody = new HashMap<>();
            requestBody.put("user", user);
            
            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.stopChatTask(taskId, requestBody);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "停止聊天任务失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("停止聊天任务失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("停止聊天任务时发生异常", e);
            return Result.error("停止聊天任务异常: " + e.getMessage());
        }
    }

    @Override
    public Result messageFeedback(String messageId, FeedbackRequest feedbackRequest) {
        try {
            Call<ResponseBody> call = httpApi.messageFeedback(messageId, feedbackRequest);
            Response<ResponseBody> response = call.execute();
            return handleResponse(response, "消息反馈");
        } catch (Exception e) {
            log.error("消息反馈时发生异常", e);
            return Result.error("消息反馈异常: " + e.getMessage());
        }
    }
    
    /**
     * 处理 Retrofit 响应
     */
    private Result handleResponse(Response<ResponseBody> response, String operationName) throws IOException {
        if (response.isSuccessful() && response.body() != null) {
            String responseBody = response.body().string();
            JsonNode jsonNode = objectMapper.readTree(responseBody);
            return Result.success(jsonNode);
        } else {
            String errorMsg = operationName + "失败";
            if (response.errorBody() != null) {
                errorMsg = response.errorBody().string();
                log.error("{}失败:{}", operationName, errorMsg);
            }
            return Result.error(errorMsg);
        }
    }

    @Override
    public Result getAppFeedbacks(Integer page, Integer limit) {
        try {
            // 构建查询参数
            Map<String, Object> params = new HashMap<>();
            params.put("page", page != null ? page : 1);
            params.put("limit", limit != null ? limit : 20);

            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.getAppFeedbacks(params);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "获取反馈失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("获取应用反馈时发生异常", e);
            return Result.error("获取反馈异常: " + e.getMessage());
        }
    }
    @Override
    public Result getSuggestedQuestions(String messageId, String user) {
        try {
            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.getSuggestedQuestions(messageId, user);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "获取建议问题失败";
                if (response.errorBody() != null) {

                    errorMsg = response.errorBody().string();
                    log.error("获取建议问题失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("获取建议问题时发生异常", e);
            return Result.error("获取建议问题异常: " + e.getMessage());
        }
    }
    @Override
    public Result getMessageHistory(MessageHistoryQuery query) {
        try {
            // 构建查询参数
            Map<String, Object> params = new HashMap<>();
            params.put("user", query.getUser());

            if (query.getConversationId() != null && !query.getConversationId().isEmpty()) {
                params.put("conversation_id", query.getConversationId());
            }

            if (query.getFirstId() != null && !query.getFirstId().isEmpty()) {
                params.put("first_id", query.getFirstId());
            }

            params.put("limit", query.getLimit() != null ? query.getLimit() : 20);

            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.getMessageHistory(params);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "获取消息历史失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("获取消息历史失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("获取消息历史时发生异常", e);
            return Result.error("获取消息历史异常: " + e.getMessage());
        }
    }

    private void enqueueSuccess(Call<ResponseBody> responseBodyCall, ResponseBodyEmitter emitter) {
        responseBodyCall.enqueue(new retrofit2.Callback<ResponseBody>() {
            @Override
            public void onResponse(Call<ResponseBody> call, Response<ResponseBody> response) {
                if (response.isSuccessful()){
                    if (response.body() != null) {

//                            emitter.send(JSONObject.parseObject(Result.success()));
                    }else {
                        throw new RuntimeException("反馈失败!");
//                            emitter.send("");
                    }
                    emitter.complete();
                }
            }

            @Override
            public void onFailure(Call<ResponseBody> call, Throwable throwable) {
                throw new RuntimeException(throwable);
            }
        });
    }

    private void enqueue(Call<ResponseBody> responseBodyCall, ResponseBodyEmitter emitter) {
        log.info("开始执行 enqueue，准备发送请求");
        long startTime = System.currentTimeMillis();

        responseBodyCall.enqueue(new retrofit2.Callback<ResponseBody>() {

            @Override
            public void onResponse(Call<ResponseBody> call, Response<ResponseBody> response) {
                long responseTime = System.currentTimeMillis() - startTime;
                log.info("收到响应，耗时: {}ms, 状态码: {}", responseTime, response.code());

                if (response.isSuccessful()) {
                    log.info("请求成功，开始处理流响应");
                    try {
                        Long starttimeo = System.currentTimeMillis();
                        processStreamResponse(response, emitter);
                        log.info("流响应处理完成");
                        log.info("总耗时: {}ms", System.currentTimeMillis()-starttimeo);
                    } catch (IOException e) {
                        log.error("处理流响应时发生IO异常", e);
                        throw new RuntimeException(e);
                    }
                } else {
                    log.info("请求失败，状态码: {}", response.code());
                    try {
                        if (response.errorBody() != null) {
                            log.info("错误信息: {}", response.errorBody().string());
                            emitter.complete();
                        }
                    } catch (IOException e) {
                        log.error("读取错误信息失败", e);
                        throw new RuntimeException(e);
                    }
                    try {
                        emitter.send(response.errorBody().string());
                    } catch (IOException e) {
                        log.error("发送错误信息失败", e);
                        throw new RuntimeException(e);
                    }
                }
            }

            @Override
            public void onFailure(Call<ResponseBody> call, Throwable throwable) {
                long failTime = System.currentTimeMillis() - startTime;
                log.error("请求失败，耗时: {}ms", failTime, throwable);
            }
        });
    }

    @Override
    public Result getConversations(ConversationQuery query) {
        try {
            // 构建查询参数
            Map<String, Object> params = new HashMap<>();
            params.put("user", query.getUser());

            if (query.getLastId() != null && !query.getLastId().isEmpty()) {
                params.put("last_id", query.getLastId());
            }

            // 限制 limit 范围在 1-100 之间
            int limit = query.getLimit() != null ? query.getLimit() : 20;
            limit = Math.max(1, Math.min(100, limit));
            params.put("limit", limit);

            if (query.getSortBy() != null && !query.getSortBy().isEmpty()) {
                params.put("sort_by", query.getSortBy());
            }

            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.getConversations(params);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "获取会话列表失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("获取会话列表失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("获取会话列表时发生异常", e);
            return Result.error("获取会话列表异常: " + e.getMessage());
        }
    }
    @Override
    public Result deleteConversation(String conversationId, String user) {
        try {
            // 创建请求对象
            DeleteConversationRequest request = new DeleteConversationRequest();
            request.setUser(user);

            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.deleteConversation(conversationId, request);
            Response<ResponseBody> response = call.execute();

            // 根据响应状态判断结果
            if (response.isSuccessful()) {
                // 204 No Content 表示删除成功
                return Result.success("success");
            } else {
                String errorMsg = "删除会话失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("删除会话失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("删除会话时发生异常", e);
            return Result.error("删除会话异常: " + e.getMessage());
        }
    }
    @Override
    public Result renameConversation(String conversationId, ConversationRenameRequest request) {
        try {
            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.renameConversation(conversationId, request);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "会话重命名失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("会话重命名失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("会话重命名时发生异常", e);
            return Result.error("会话重命名异常: " + e.getMessage());
        }
    }
    @Override
    public Result getConversationVariables(String conversationId, ConversationVariablesQuery query) {
        try {
            // 构建查询参数
            Map<String, Object> params = new HashMap<>();
            params.put("user", query.getUser());

            if (query.getLastId() != null && !query.getLastId().isEmpty()) {
                params.put("last_id", query.getLastId());
            }

            // 限制 limit 范围在 1-100 之间
            int limit = query.getLimit() != null ? query.getLimit() : 20;
            limit = Math.max(1, Math.min(100, limit));
            params.put("limit", limit);

            if (query.getVariableName() != null && !query.getVariableName().isEmpty()) {
                params.put("variable_name", query.getVariableName());
            }

            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.getConversationVariables(conversationId, params);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "获取对话变量失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("获取对话变量失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("获取对话变量时发生异常", e);
            return Result.error("获取对话变量异常: " + e.getMessage());
        }
    }
    @Override
    public Result audioToText(AudioToTextRequest request) {
        try {
            // 创建 multipart 请求体
            RequestBody userRequestBody = RequestBody.create(okhttp3.MediaType.parse("text/plain"), request.getUser());

            // 创建文件 part
            okhttp3.RequestBody fileBody = okhttp3.RequestBody.create(
                    okhttp3.MediaType.parse("audio/*"),
                    request.getFile().getBytes()
            );
            MultipartBody.Part filePart = MultipartBody.Part.createFormData(
                    "file",
                    request.getFile().getOriginalFilename(),
                    fileBody
            );

            // 调用 Retrofit 接口
            Call<ResponseBody> call = uploadHttpApi.audioToText(filePart, userRequestBody);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "语音转文字失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("语音转文字失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("语音转文字时发生异常", e);
            return Result.error("语音转文字异常: " + e.getMessage());
        }
    }
    @Override
    public Result textToAudio(TextToAudioRequest request) {
        try {
            // 调用 Retrofit 接口
            Call<ResponseBody> call = audioApi.textToAudio(request);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 获取响应体的字节数据
                ResponseBody responseBody = response.body();
                byte[] audioData = responseBody.bytes();

                // 创建包含音频数据的结果
                Map<String, Object> result = new HashMap<>();
                result.put("audioData", audioData);
                result.put("contentType", response.headers().get("Content-Type"));

                return Result.success(result);
            } else {
                String errorMsg = "文字转语音失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("文字转语音失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("文字转语音时发生异常", e);
            return Result.error("文字转语音异常: " + e.getMessage());
        }
    }
    @Override
    public Result getAppInfo() {
        try {
            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.getAppInfo();
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "获取应用信息失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("获取应用信息失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("获取应用信息时发生异常", e);
            return Result.error("获取应用信息异常: " + e.getMessage());
        }
    }
    @Override
    public Result getAppMeta() {
        try {
            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.getAppMeta();
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "获取应用Meta信息失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("获取应用Meta信息失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("获取应用Meta信息时发生异常", e);
            return Result.error("获取应用Meta信息异常: " + e.getMessage());
        }
    }
    @Override
    public Result getWebAppSettings() {
        try {
            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.getWebAppSettings();
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "获取WebApp设置失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("获取WebApp设置失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("获取WebApp设置时发生异常", e);
            return Result.error("获取WebApp设置异常: " + e.getMessage());
        }
    }
    
    @Override
    public Result getAnnotations(Integer page, Integer limit) {
        try {
            // 构建查询参数
            Map<String, Object> params = new HashMap<>();
            params.put("page", page != null ? page : 1);
            params.put("limit", limit != null ? limit : 20);

            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.getAnnotations(params);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "获取标注列表失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("获取标注列表失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("获取标注列表时发生异常", e);
            return Result.error("获取标注列表异常: " + e.getMessage());
        }
    }
    
    @Override
    public Result createAnnotation(CreateAnnotationRequest request) {
        try {
            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.createAnnotation(request);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "创建标注失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("创建标注失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("创建标注时发生异常", e);
            return Result.error("创建标注异常: " + e.getMessage());
        }
    }
    
    @Override
    public Result updateAnnotation(String annotationId, UpdateAnnotationRequest request) {
        try {
            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.updateAnnotation(annotationId, request);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "更新标注失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("更新标注失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("更新标注时发生异常", e);
            return Result.error("更新标注异常: " + e.getMessage());
        }
    }
    
    @Override
    public Result deleteAnnotation(String annotationId) {
        try {
            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.deleteAnnotation(annotationId);
            Response<ResponseBody> response = call.execute();

            // 根据响应状态判断结果
            if (response.isSuccessful()) {
                // 204 No Content 表示删除成功
                return Result.success("success");
            } else {
                String errorMsg = "删除标注失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("删除标注失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("删除标注时发生异常", e);
            return Result.error("删除标注异常: " + e.getMessage());
        }
    }
    
    @Override
    public Result annotationReply(String action, AnnotationReplyRequest request) {
        try {
            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.annotationReply(action, request);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "标注回复设置失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("标注回复设置失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("标注回复设置时发生异常", e);
            return Result.error("标注回复设置异常: " + e.getMessage());
        }
    }
    
    @Override
    public Result getAnnotationReplyStatus(String action, String jobId) {
        try {
            // 调用 Retrofit 接口
            Call<ResponseBody> call = httpApi.getAnnotationReplyStatus(action, jobId);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "查询标注回复任务状态失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("查询标注回复任务状态失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("查询标注回复任务状态时发生异常", e);
            return Result.error("查询标注回复任务状态异常: " + e.getMessage());
        }
    }

    @Override
    public Result uploadFile(MultipartFile file, String user) {
        try {
            // 创建 MultipartBody.Part
            RequestBody requestFile = RequestBody.create(okhttp3.MediaType.parse("application/octet-stream"), file.getBytes());
            MultipartBody.Part filePart = MultipartBody.Part.createFormData("file", file.getOriginalFilename(), requestFile);
            
            // 创建 user 参数
            RequestBody userBody = RequestBody.create(okhttp3.MediaType.parse("text/plain"), user);
            
            // 调用 Retrofit 接口
            Call<ResponseBody> call = uploadHttpApi.uploadFile(filePart, userBody);
            Response<ResponseBody> response = call.execute();

            if (response.isSuccessful() && response.body() != null) {
                // 解析响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);

                // 返回成功结果
                return Result.success(jsonNode);
            } else {
                String errorMsg = "文件上传失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                    log.error("文件上传失败:{}", errorMsg);
                }
                return Result.error(errorMsg);
            }
        } catch (Exception e) {
            log.error("文件上传时发生异常", e);
            return Result.error("文件上传异常: " + e.getMessage());
        }
    }

    private void sendDifyRequest(DifyRequest difyRequest, ResponseBodyEmitter emitter) {
        try {
            // 构建请求 JSON
            String requestBody = objectMapper.writeValueAsString(difyRequest);
            log.info("发送到 Dify 的请求: {}", requestBody);

            // 发送 HTTP 请求
            HttpResponse response = HttpRequest.post(diFyConfig.getBaseUrl()+"chat-messages")
                    .header("Authorization", "Bearer " + diFyConfig.getApiKey())
                    .header("Content-Type", "application/json")
                    .header("Accept", "text/event-stream")
                    .body(requestBody)
                    .executeAsync();
            log.info("访问路径{}", diFyConfig.getBaseUrl()+"chat-messages");
            // 处理响应流
            if (response.getStatus() == 200) {
//                processStreamResponse(response, emitter);
            } else {
                String errorMsg = "Dify API 返回错误状态码: " + response.getStatus()
                        + ", 响应内容: " + response.body();
                log.error(errorMsg);
                emitter.send(errorMsg);
            }
        } catch (Exception e) {
            log.error("发送 Dify 请求时发生异常", e);
            try {
                emitter.send("请求异常: " + e.getMessage());
            } catch (IOException ioException) {
                log.error("发送异常信息失败", ioException);
            }
        } finally {
            emitter.complete();
        }
    }

    private void processStreamResponse(Response<ResponseBody> response, ResponseBodyEmitter emitter) throws IOException {
        try {
            ResponseBody responseBody = response.body();
            if (responseBody != null) {
                BufferedReader reader = new BufferedReader(new InputStreamReader(responseBody.byteStream()));
                String line;
                String taskId = null; // 用于存储 task_id
                String messageId = null;

                while ((line = reader.readLine()) != null) {
                    // 处理 SSE 数据
                    if (line.startsWith("data: ")) {
                        String jsonData = line.substring(6).trim();
                        try {
                            // 解析 JSON 数据
                            JsonNode jsonNode = objectMapper.readTree(jsonData);
                            String eventType = jsonNode.get("event").asText();
                            
                            // 提取 task_id（通常在 message 或 agent_message 事件中）
                            if (jsonNode.has("task_id") && taskId == null) {
                                taskId = jsonNode.get("task_id").asText();
                                log.info("任务ID task_id: {}", taskId);
                            }
                            
                            if (jsonNode.has("message_id") && messageId == null){
                                messageId = jsonNode.get("message_id").asText();
                                log.info("消息ID message_id: {}", messageId);
                            }
                            
                            // 根据事件类型处理数据
                            switch (eventType) {
                                case "message":
                                case "agent_message":
                                    String answer = jsonNode.get("answer").asText();
                                    
                                    // 使用正则表达式过滤掉think标签及其内容
                                    answer = answer.replaceAll("<think>[\\s\\S]*?</think>", "");
                                    
                                    // 发送答案给客户端
                                    if (answer != null && !answer.isEmpty()) {
                                        emitter.send(answer.trim());
                                    }
                                    break;
                                    
                                case "message_end":
                                    // 消息结束，可以获取 conversation_id 等信息
                                    String conversationId = jsonNode.get("conversation_id").asText();
                                    log.info("会话ID conversation_id: {}", conversationId);
                                    
//                                    // 构建结束响应对象
//                                    Map<String, Object> endResponse = new HashMap<>();
//                                    endResponse.put("type", "message_end");
//                                    endResponse.put("conversation_id", conversationId);
                                    
//                                    // 添加 message_id（如果存在）
//                                    if (messageId != null) {
//                                        endResponse.put("message_id", messageId);
//                                    }
//
//                                    // 发送消息结束通知
//                                    emitter.send(JSONUtil.toJsonStr(endResponse));
                                    break;
                                    
                                case "ping":
                                    // 心跳包处理
                                    log.debug("收到心跳包: {}", jsonNode);
                                    
                                    // 发送心跳响应
//                                    Map<String, String> pingResponse = new HashMap<>();
//                                    pingResponse.put("type", "ping");
//                                    pingResponse.put("timestamp", String.valueOf(System.currentTimeMillis()));
//                                    emitter.send(JSONUtil.toJsonStr(pingResponse));
                                    break;
                                    
                                case "error":
                                    // 错误事件处理
                                    String errorMessage = jsonNode.get("error").get("message").asText();
                                    log.error("收到错误事件: {}", errorMessage);
                                    
//                                    // 构建错误响应对象
//                                    Map<String, String> errorResponse = new HashMap<>();
//                                    errorResponse.put("type", "error");
//                                    errorResponse.put("message", errorMessage);
                                    
                                    // 发送错误信息给客户端
//                                    emitter.send(JSONUtil.toJsonStr(errorResponse));
                                    break;
                                    
                                case "workflow_started":
                                    // 工作流开始事件
                                    log.info("工作流开始: {}", jsonNode);
                                    
                                    // 构建工作流开始响应
//                                    Map<String, Object> workflowStartResponse = new HashMap<>();
//                                    workflowStartResponse.put("type", "workflow_started");
//                                    workflowStartResponse.put("workflow_id", jsonNode.get("data").get("workflow_id").asText());
                                    
                                    // 发送工作流开始通知
//                                    emitter.send(JSONUtil.toJsonStr(workflowStartResponse));
                                    break;
                                    
                                case "workflow_finished":
                                    // 工作流结束事件
                                    log.info("工作流结束: {}", jsonNode);
                                    
                                    // 构建工作流结束响应
//                                    Map<String, Object> workflowEndResponse = new HashMap<>();
//                                    workflowEndResponse.put("type", "workflow_finished");
//                                    workflowEndResponse.put("data", jsonNode.get("data"));
//
//                                    // 添加 workflow_run_id（如果存在）
//                                    if (jsonNode.has("workflow_run_id")) {
//                                        workflowEndResponse.put("workflow_run_id", jsonNode.get("workflow_run_id").asText());
//                                    }
                                    
                                    // 发送工作流结束通知
//                                    emitter.send(JSONUtil.toJsonStr(workflowEndResponse));
                                    break;
                                    
                                case "annotation_reply":
                                    // 标注回复事件
                                    log.debug("收到标注回复事件: {}", jsonNode);
                                    
//                                    // 构建标注回复响应
//                                    Map<String, Object> annotationResponse = new HashMap<>();
//                                    annotationResponse.put("type", "annotation_reply");
//                                    annotationResponse.put("data", jsonNode);
//
//                                    // 发送标注回复通知
//                                    emitter.send(JSONUtil.toJsonStr(annotationResponse));
                                    break;
                                    
                                default:
                                    log.debug("收到未处理的事件类型: {}", eventType);
                                    
//                                    // 发送原始事件数据
//                                    Map<String, Object> rawEvent = new HashMap<>();
//                                    rawEvent.put("type", "raw");
//                                    rawEvent.put("event", eventType);
//                                    rawEvent.put("data", jsonNode);
//                                    emitter.send(JSONUtil.toJsonStr(rawEvent));
                                    break;
                            }
                        } catch (Exception e) {
                            log.warn("解析 JSON 数据失败: {}", jsonData, e);
                            // 发送原始数据作为回退
//                                Map<String, String> rawData = new HashMap<>();
//                                rawData.put("type", "raw_data");
//                                rawData.put("content", jsonData);
//                                emitter.send(JSONUtil.toJsonStr(rawData));
                        }
                    } else if (line.startsWith("event: ")) {
                        // 事件类型行，可以记录但不需要发送给客户端
                        log.debug("收到事件类型: {}", line.substring(7));
                    } else if (line.trim().isEmpty()) {
                        // 空行，SSE 格式的分隔符
                        continue;
                    } else {
                        // 其他行，可能是错误信息或其他数据
                        log.debug("收到其他数据: {}", line);
                    }
                }
            }
        } catch (Exception e) {
            log.error("处理流响应时发生异常", e);
            try {
                emitter.send("处理响应异常: " + e.getMessage());
            } catch (IOException ioException) {
                log.error("发送异常信息失败", ioException);
            }
        } finally {
            try {
                emitter.complete();
            } catch (Exception e) {
                log.warn("完成 emitter 时发生异常", e);
            }
        }
    }
    private void processStreamResponse(HttpResponse response, ResponseBodyEmitter emitter) {
        try {
            if (response.getStatus() == 200) {
                // 获取响应流
                InputStream inputStream = response.bodyStream();
                BufferedReader reader = new BufferedReader(new InputStreamReader(inputStream));
                String line;
                String taskId = null;

                while ((line = reader.readLine()) != null) {
                    // 立即发送每一行数据以实现流式传输
                    try {
                        emitter.send(line + "\n");
                    } catch (IOException e) {
                        log.warn("发送数据到客户端失败，可能客户端已断开连接", e);
                        break; // 客户端断开连接，停止处理
                    }

                    // 处理 SSE 数据以提取 task_id 等信息
                    if (line.startsWith("data: ")) {
                        String jsonData = line.substring(6);
                        try {
                            // 解析 JSON 数据
                            JsonNode jsonNode = objectMapper.readTree(jsonData);
                            String eventType = jsonNode.get("event").asText();

                            // 提取 task_id（通常在 message 或 agent_message 事件中）
                            if (jsonNode.has("task_id") && taskId == null) {
                                taskId = jsonNode.get("task_id").asText();
                                log.info("获取到 task_id: {}", taskId);
                            }

                            // 根据事件类型处理数据（这部分数据已经在上面发送了）
                            switch (eventType) {
                                case "message":
                                case "agent_message":
                                    // 数据内容已经在 emitter.send(line + "\n") 中发送
                                    break;
                                case "message_end":
                                    String conversationId = jsonNode.get("conversation_id").asText();
                                    log.info("会话ID: {}", conversationId);
                                    break;
                                case "ping":
                                    // 心跳包处理
                                    log.debug("收到心跳包:{}", line);
                                    break;
                                default:
                                    log.debug("收到未处理的事件类型: {}", eventType);
                                    break;
                            }
                        } catch (Exception e) {
                            log.warn("解析 JSON 数据失败: {}", jsonData, e);
                        }
                    } else if (line.startsWith("event: ")) {
                        log.debug("收到事件类型: {}", line.substring(7));
                    } else if (line.trim().isEmpty()) {
                        continue;
                    } else {
                        log.debug("收到其他数据: {}", line);
                    }
                }
            } else {
                String errorMsg = "Dify API 返回错误状态码: " + response.getStatus()
                        + ", 响应内容: " + response.body();
                log.error(errorMsg);
                emitter.send(errorMsg);
            }
        } catch (Exception e) {
            log.error("处理流响应时发生异常", e);
            try {
                emitter.send("处理响应异常: " + e.getMessage());
            } catch (IOException ioException) {
                log.error("发送异常信息失败", ioException);
            }
        } finally {
            try {
                emitter.complete();
            } catch (Exception e) {
                log.warn("完成 emitter 时发生异常", e);
            }
        }
    }

    /**
     * 处理阻塞模式响应
     * @param response Dify API响应
     * @return 处理后的响应结果
     */
    private Result handleBlockingResponse(Response<ResponseBody> response) {
        try {
            if (response.isSuccessful() && response.body() != null) {
                // 获取响应体
                String responseBody = response.body().string();
                JsonNode jsonNode = objectMapper.readTree(responseBody);
                
                // 如果是message事件且包含answer字段，则过滤think标签
                if (jsonNode != null && jsonNode.has("event") && 
                    ("message".equals(jsonNode.get("event").asText()) || 
                     "agent_message".equals(jsonNode.get("event").asText())) && 
                    jsonNode.has("answer")) {
                    
                    String answer = jsonNode.get("answer").asText();
                    // 使用正则表达式过滤掉think标签及其内容
                    answer = answer.replaceAll("<tool_call>[\\s\\S]*?</tool_call>", "");
                    
                    // 创建新的结果对象，避免直接修改原始JSON结构
                    Map<String, Object> result = new HashMap<>();
                    result.put("event", jsonNode.get("event").asText());
                    result.put("answer", answer);
                    
                    // 如果存在message_id，也包含进去
                    if (jsonNode.has("message_id")) {
                        result.put("message_id", jsonNode.get("message_id").asText());
                    }
                    
                    // 如果存在conversation_id，也包含进去
                    if (jsonNode.has("conversation_id")) {
                        result.put("conversation_id", jsonNode.get("conversation_id").asText());
                    }
                    
                    // 如果存在其他字段，也包含进去
                    if (jsonNode.has("task_id")) {
                        result.put("task_id", jsonNode.get("task_id").asText());
                    }
                    
                    if (jsonNode.has("id")) {
                        result.put("id", jsonNode.get("id").asText());
                    }
                    
                    if (jsonNode.has("mode")) {
                        result.put("mode", jsonNode.get("mode").asText());
                    }
                    
                    if (jsonNode.has("created_at")) {
                        result.put("created_at", jsonNode.get("created_at").asLong());
                    }
                    
                    // 如果存在metadata，也包含进去
                    if (jsonNode.has("metadata")) {
                        result.put("metadata", jsonNode.get("metadata"));
                    }
                    
                    return Result.success(result);
                }
                return Result.success(jsonNode);
            } else {
                String errorMsg = "请求失败";
                if (response.errorBody() != null) {
                    errorMsg = response.errorBody().string();
                }
                return Result.error(errorMsg);
            }
        } catch (JsonProcessingException jsonEx) {
            log.error("JSON 解析失败", jsonEx);
            return Result.error("JSON 解析失败: " + jsonEx.getMessage());
        } catch (IOException ioEx) {
            log.error("读取响应流失败", ioEx);
            return Result.error("读取响应流失败: " + ioEx.getMessage());
        } catch (Exception e) {
            log.error("处理阻塞模式响应时发生异常", e);
            return Result.error("处理阻塞模式响应异常: " + e.getMessage());
        }
    }
}
